package com.chiaching.pms.msg;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.chiaching.framwork.domian.constant.Action;
import com.chiaching.framwork.domian.constant.RocketTag;
import com.chiaching.framwork.domian.constant.RocketTopic;
import com.chiaching.framwork.utils.RocketUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.stream.Stream;

/**
 * @author jiajing.he
 * @date 2021/4/11 11:34
 */
@Component
@Slf4j
public class DefaultMqListener implements ApplicationRunner {

    @Autowired
    private RocketUtils rocketUtils;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Stream.of(RocketTopic.values()).forEach(m ->{
            rocketUtils.subscribe(m.getName(),RocketTag.ALL,(e) ->{
                try {
                    String body = new String(e.getBody(), RemotingHelper.DEFAULT_CHARSET);
                    log.info("topic:[{}] tag:[{}]  监听到消息:{}",e.getTopic(),e.getTags(),body);
                    switch (RocketTopic.valueOf(e.getTopic())){
                        case SUPER_MALL_CANAL:consumerSuperMallCanal(e,body);break;
                        case SUPER_MALL_COMMON:consumerSuperMallCommon(e);break;
                        default:break;
                    }
                } catch (UnsupportedEncodingException unsupportedEncodingException) {
                    log.error("解析mq消息体异常  param[ topic:{},tag:{} ]  ", e.getTopic(),e.getTags(),unsupportedEncodingException);
                   return Action.RECONSUME_LATER;
                }
                return Action.CONSUME_SUCCESS;
            });
        });

    }


    private void consumerSuperMallCanal(MessageExt messageExt,String body){
        switch (messageExt.getTags()){
            case RocketTag.DB_CHANGE:dbChangeHandler(messageExt,body);break;
            default:break;
        }
    }

    private void dbChangeHandler(MessageExt messageExt,String body) {
        JSONObject jsonObject = JSONObject.parseObject(body);
        //jsonObject;
        //CanalEntry.Entry

    }

    private void consumerSuperMallCommon(MessageExt messageExt){

    }
}
